package batch;

import org.apache.commons.collections.CollectionUtils;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.MQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * 把消息list拆分,进行批量消息发送
 * @author: BacHe
 * @email: 1218585258@qq.com
 * @Date: 2020/12/10 23:24
 */
public class ListSplitter implements Iterator<List<Message>> {
    //不能超过4M,（ 1024 * 1024 * 4）否则会抛异常：CODE: 13  DESC: the message body size over max value, MAX: 4194304
    //由于这里遍历统计的大小只是消息的估值，此数字要偏小 （1024 * 1024 * 1），避免抛异常。
    //发送的数据越大，客户端等待响应的时间会越长，可能会超时。
    private final int SIZE_LIMIT = 1024 * 1024 * 1;
    //要拆分发送的批量数据
    private final List<Message> messages;
    //当前位置
    private int currIndex;

    private MQProducer producer;

    public ListSplitter(List<Message> messages, MQProducer producer) {
        this.messages = messages;
        this.producer = producer;
    }

    @Override
    public boolean hasNext() {
        return currIndex < messages.size();
    }

    @Override
    public List<Message> next() {
        int nextIndex = this.currIndex;
        int totalSize = 0;
        for (; nextIndex < messages.size(); nextIndex++) {
            Message message = messages.get(nextIndex);
            int tmpSize = message.getTopic().length() + message.getBody().length;
            Map<String, String> properties = message.getProperties();
            for (Map.Entry<String, String> entry : properties.entrySet()) {
                tmpSize += entry.getKey().length() + entry.getValue().length();
            }
            //增加日志开销20字节
            tmpSize = tmpSize + 20;
            if (tmpSize > SIZE_LIMIT) {
                //单个消息超过最大的限制
                //忽略，否则会阻塞分裂的进程
                if (nextIndex-currIndex==0){
                    //假如下一个子列表没有元素，则添加这个子列表然后退出循环，
                    // 否则只是退出循环
                    nextIndex++;
                }
                break;
            }
            if (tmpSize + totalSize > SIZE_LIMIT) {
                break;
            } else {
                totalSize += tmpSize;
            }
        }
        //截取当前位置，到下一个位置。
        List<Message> subList = messages.subList(currIndex, nextIndex);
        //下一个位置，复制给当前位置。
        currIndex = nextIndex;
        return subList;
    }
    //把消息拆分并发送。
    public void send() throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        while (this.hasNext()) {
            List<Message> messageList = this.next();
            if (CollectionUtils.isNotEmpty(messageList)) {
                SendResult send = producer.send(messageList);
                System.out.println("SendResult:" + send.getSendStatus());
            }
        }
    }
}
